package server;

import common.SynUtils;
import hdfs.hdfsPut;
import com.google.protobuf.InvalidProtocolBufferException;
import common.BlockInfo;
import common.DNInfo;
import common.FileInfo;
import interfaces.DataNodeDef;
import interfaces.NameNodeDef;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import static common.StateCode.*;
import static common.Utils.*;

public class NameNode extends UnicastRemoteObject implements  NameNodeDef {
    // 不是 static 可能有问题，需要测试
    private Map<String, FileInfo> fileMap = new ConcurrentHashMap<>();
    private Map<Integer, BlockInfo> blockMap = new ConcurrentHashMap<>();
    private List<DNInfo> dNList = new LinkedList<>();
    private Integer nextBlock = 1;
    private static final Logger logger = LoggerFactory.getLogger(NameNode.class);
    private Map<String,Date> DNListener = new ConcurrentHashMap<>();

    @Override
    public byte[] SynFile(byte[] inp) throws RemoteException {
        hdfsPut.SynFileRequest synFileRequest = null;
        try {
            synFileRequest = hdfsPut.SynFileRequest.parseFrom(inp);
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        assert synFileRequest != null;
        String filename = synFileRequest.getFilename();
        FileInfo fileInfo = fileMap.get(filename);
        hdfsPut.SynFileResponse.Builder builder = hdfsPut.SynFileResponse.newBuilder();
        try {
            SynUtils.writeFileInfo(fileInfo);
            for(Integer i:blockMap.keySet()){
                BlockInfo blockInfo = blockMap.get(i);
                if(blockInfo.getFileName().equals(filename)){
                    SynUtils.writeBlockMap(blockInfo);
                }
            }
            builder.setStatus(fileOk);
            logger.info(filename+"fileinfo,blockmap写入磁盘成功");
        } catch (IOException e) {
            logger.error(filename+"fileinfo,blockmap写入磁盘失败");
            e.printStackTrace();
        }
        return builder.build().toByteArray();
    }

    @Override
    public byte[] getHeartBeat(byte[] inp) throws RemoteException{
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                if (DNListener.size() == 0) {
                    for (DNInfo dnInfo : dNList) {
                        DNListener.put(dnInfo.getIp() + "," + dnInfo.getPort(), new Date());
                    }
                }
                for (String s : DNListener.keySet()) {
                    String[] split = s.split(",");
                    String IP = split[0];
                    int PORT = Integer.parseInt(split[1]);
                    DataNodeDef dnStub = getDNStub(IP, PORT);
                    byte[] bytes = hdfsPut.heartBeatRequest.newBuilder().setPING("PING").build().toByteArray();
                    if(dnStub==null){
                        continue;
                    }
                    hdfsPut.heartBeatResponse heartBeatResponse = null;
                    try {
                        heartBeatResponse = hdfsPut.heartBeatResponse.parseFrom(dnStub.heartBeat(bytes));
                        if (heartBeatResponse.getPONG().equals("PONG")) {
                            DNListener.put(s, new Date());
                            logger.info("收到"+"dataNode"+IP+PORT+"心跳");
                        }
                    } catch (InvalidProtocolBufferException | RemoteException e) {
                        e.printStackTrace();
                    }
                }
            };
        },1000*60,1000*100);  //延迟1分钟开始，每100秒进行一次
        return new byte[0];
    }

    public void heartBeatListener(){
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                for(String s:DNListener.keySet()){
                    long time = new Date().getTime();
                    if(time-DNListener.get(s).getTime()>1000*1200){
                        String[] split = s.split(",");
                        String IP = split[0];
                        int PORT = Integer.parseInt(split[1]);
                        for(DNInfo dnInfo:dNList){
                            if(dnInfo.getIp().equals(IP) && dnInfo.getPort()==PORT){
                                dnInfo.setAlive(false);
                                logger.info(IP+","+PORT+"DATANODE失去连接");
                            }
                        }
                    }
                }
            }
        },1000*60,1000*300);
    }

    public NameNode() throws RemoteException {
        super();
        init();
        heartBeatListener();
        try {
            getHeartBeat(new byte[0]);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public void init(){
        //判断，如果文件为空的话就进行下面的默认活动
        File file = new File("dNList");
        if(file.exists()){
            //从文件中读数据
            try {
                List<DNInfo> dataNodeList = SynUtils.readdNList();
                dNList.addAll(dataNodeList);
                logger.info("读取datanode数据成功"+",共"+dNList.size()+"个");
                List<BlockInfo> blockInfoList = SynUtils.readBlockMap();
                for (BlockInfo blockInfo : blockInfoList) {
                    blockMap.put(blockInfo.getBlockNO(),blockInfo);
                }
                logger.info("读取blockInfo数据成功");
                List<FileInfo> fileInfos = SynUtils.readFileInfo();
                for (FileInfo fileInfo : fileInfos) {
                    fileMap.put(fileInfo.getFileName(),fileInfo);
                }
                logger.info("读取fileInfo数据成功");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }else{
            dNList.add(new DNInfo(remoteIp,startPort,startTPort));
            dNList.add(new DNInfo(remoteIp,startPort+1,startTPort+1));
            dNList.add(new DNInfo(remoteIp,startPort+2,startTPort+2));
            try {
                SynUtils.writedNList(new DNInfo(remoteIp,startPort,startTPort));
                SynUtils.writedNList(new DNInfo(remoteIp,startPort+1,startTPort+1));
                SynUtils.writedNList(new DNInfo(remoteIp,startPort+2,startTPort+2));
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }

    /**
     *
     * @param inp openFileRequest
     * @return 如果是写请求，如果文件已存在且状态OK，则返回文件存在，如果不存在则在fileMap中新增加文件信息，返回fileOk
     *如果是读请求，若文件不存在或者状态错误返回fileError，否则从fileInfo中读取块信息，返回块信息
     * @throws RemoteException
     */
    @Override
    public byte[] openFile(byte[] inp) throws RemoteException {
        hdfsPut.OpenFileResponse.Builder response = hdfsPut.OpenFileResponse.newBuilder();
        try {
            hdfsPut.OpenFileRequest req = hdfsPut.OpenFileRequest.parseFrom(inp);
            String filename = req.getFileName();
            boolean forRead = req.getForRead();
            if(!forRead){       //如果是写请求
                if(fileMap.containsKey(filename)&&fileMap.get(filename).getStatus()==fileOk){
                    response.setStatus(fileExist);
                    return response.build().toByteArray();
                }
                fileMap.put(filename,new FileInfo(filename));
                // 可以上传
                response.setStatus(fileOk);
            }else{      //如果是读请求
                if(!fileMap.containsKey(filename)){
                    response.setStatus(fileError);
                    logger.error("文件"+filename+"不存在");
                    return response.build().toByteArray();
                }
                FileInfo fileInfo = fileMap.get(filename);
                if(fileInfo.getStatus()!= fileOk){
                    response.setStatus(fileError);
                    logger.error("文件"+filename+"状态错误");
                    return response.build().toByteArray();
                }
                response.setStatus(fileOk);
                List<Integer> blockList = fileInfo.getBlockList();
                response.addAllBlockNums(blockList);
            }
            return response.build().toByteArray();

        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public byte[] closeFile(byte[] inp) throws RemoteException {
        try {
            hdfsPut.CloseFileRequest closeReq = hdfsPut.CloseFileRequest.parseFrom(inp);
            String fileName = closeReq.getFileName();
            fileMap.get(fileName).setStatus(fileOk);
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return hdfsPut.CloseFileResponse.newBuilder().setStatus(fileOk).build().toByteArray();
    }

    /**
     *
     * @param inp getBlockLocationsRequest
     * @return 通过请求中的块序号从blockMap中获取位置信息，即List<DNInfo>，进一步得到ip和port，返回状态1,如果从blockMap中得不到位置信息
     * 则返回状态400
     * @throws RemoteException
     */
    @Override
    public byte[] getBlockLocations(byte[] inp) throws RemoteException {
        try {
            hdfsPut.blockLocationResponse.Builder blockResponse = hdfsPut.blockLocationResponse.newBuilder();
            hdfsPut.blockLocationRequest blockRequest = hdfsPut.blockLocationRequest.parseFrom(inp);
            int blockNum = blockRequest.getBlockNum();
            BlockInfo blockInfo = blockMap.get(blockNum);
            if(blockInfo==null){
                blockResponse.setStatus(blockError);
                return blockResponse.build().toByteArray();
            }
            List<DNInfo> locations = blockInfo.getLocations();
            for(DNInfo dnInfo:locations){
                hdfsPut.DataNodeLocation.Builder dataLocation = hdfsPut.DataNodeLocation.newBuilder();
                dataLocation.setIp(dnInfo.getIp());
                dataLocation.setPort(dnInfo.getPort());
                blockResponse.addDataNodeLocation(dataLocation);
            }
            blockResponse.setStatus(fileOk);
            return blockResponse.build().toByteArray();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    @Override
    public byte[] assignBlock(byte[] inp) throws RemoteException {
        try {
            hdfsPut.AssignBlockRequest req = hdfsPut.AssignBlockRequest.parseFrom(inp);
            String filename = req.getFilename();
            hdfsPut.AssignBlockResponse.Builder response = hdfsPut.AssignBlockResponse.newBuilder();
            hdfsPut.BlockLocations.Builder locations = hdfsPut.BlockLocations.newBuilder();
            int fileSize = req.getSize();
            // 创建空块
            synchronized (nextBlock){
                FileInfo fileInfo = fileMap.get(filename);
                fileInfo.appendBlock(nextBlock);
                blockMap.put(nextBlock,new BlockInfo(filename,nextBlock,req.getSize()));
                locations.setBlockNumber(nextBlock);
                // 获得块的存储位置
                List<Integer> dnIdx = assignDN();
                // 组装返回信息
                for (Integer idx : dnIdx) {
                    hdfsPut.DataNodeLocation.Builder dnLocation = hdfsPut.DataNodeLocation.newBuilder();
                    DNInfo dnInfo = dNList.get(idx);
                    dnLocation.setIp(dnInfo.getIp());
                    dnLocation.setPort(dnInfo.getPort());
                    dnLocation.setTPort(dnInfo.gettPort());
                    locations.addLocations(dnLocation.build());
                    BlockInfo blockInfo = blockMap.get(nextBlock);
                    blockInfo.appendLocation(dnInfo.getIp(),dnInfo.getPort());
                    blockInfo.setValidLen(fileSize);
                }
                nextBlock++;
            }
            response.setNewBlock(locations.build());
            return response.build().toByteArray();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return new byte[0];
    }

    @Override
    public byte[] list(byte[] inp) throws RemoteException {
        hdfsPut.AllFileResponse.Builder builder = hdfsPut.AllFileResponse.newBuilder();
        for(String fileName:fileMap.keySet()){
            hdfsPut.FileInfo.Builder fileInfoBuilder = hdfsPut.FileInfo.newBuilder();
            FileInfo fileInfo = fileMap.get(fileName);
            fileInfoBuilder.setFileName(fileName)
                    .setFileStatus(fileOk)
                    .addAllBlockNum(fileInfo.getBlockList());
            builder.addFileInfo(fileInfoBuilder);
        }
        builder.setStatus(fileOk);
        return builder.build().toByteArray();
    }


    public byte[] blockReport(byte[] inp) throws RemoteException {
        try {
            hdfsPut.BlockReportRequest request = hdfsPut.BlockReportRequest.parseFrom(inp);
            String IP = request.getIP();
            int PORT = request.getPORT();
            boolean isDataNode = false;
            //检验dNList里面有没有该node，如果没有则添加,如果状态为死亡则修改为活
            //这里是状态为死亡
            for (DNInfo dnInfo : dNList) {
                if(IP.equals(dnInfo.getIp()) && PORT==dnInfo.getPort()){
                    isDataNode=true;
                    dnInfo.setAlive(true);
                    break;
                }
            }
            hdfsPut.BlockReportResponse.Builder response = hdfsPut.BlockReportResponse.newBuilder();
            if(isDataNode){
                for (hdfsPut.BlockInfo blockInfo : request.getBlocksList()) {
                    if(blockInfo.getStatus()!=1){
                        hdfsPut.ReadBlockRequest.Builder builder = hdfsPut.ReadBlockRequest.newBuilder();
                        //取出一个dataNode传输block
                        DNInfo dnInfo=dNList.get(0);
                        for (DNInfo info : dNList) {
                            if(!IP.equals(info.getIp()) || PORT!=info.getPort()){
                                dnInfo=info;
                                break;
                            }
                        }
                        logger.info("从"+dnInfo.getIp()+dnInfo.getPort()+"获取"+blockInfo.getBlockNum());
                        builder.setBlockNo(blockInfo.getBlockNum())
                                .setIP(dnInfo.getIp())
                                .setPORT(dnInfo.getPort());
                        response.addReadBlockRequest(builder);
                    }
                }
            }else{   //新增加dataNode
                dNList.add(new DNInfo(IP,PORT));
                try {
                    SynUtils.writedNList(new DNInfo(IP,PORT));
                } catch (IOException e) {
                    e.printStackTrace();
                }
                logger.info("新增加datanode "+IP+":"+PORT);
            }
            return response.build().toByteArray();
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
        return new byte[0];
    }


    /**
     *
     * @return 根据副本数返回DataNode的List
     */
    private List<Integer> assignDN() {
        final Random random = new Random();
        final Set<Integer> DNSet = new HashSet<>();
        while (DNSet.size() < REPLICATE_NUM) {
            DNSet.add(random.nextInt(dNList.size()));
        }
        return new ArrayList<>(DNSet);
    }
}
